package cn.doitedu.demo11;


import cn.doitedu.pojo.UserEvent;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import groovy.lang.GroovyClassLoader;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.RandomUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.roaringbitmap.longlong.Roaring64Bitmap;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/9/21
 * @Desc: 学大数据，上多易教育
 * <p>
 * 实现了：支持跨越规则注入时间点前后时段动态画像统计条件的规则模型
 **/

@Slf4j
public class Demo11 {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 配置一些必选参数
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.setParallelism(1);

        // 构建一个kafka source
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setTopics("user-events-log")
                .setBootstrapServers("doitedu:9092")
                .setGroupId("g001")
                //.setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.LATEST))
                .setStartingOffsets(OffsetsInitializer.latest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        // 将source添加到env，得到数据流
        DataStreamSource<String> jsonStream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "s1");

        // json解析，封装成javaBean
        SingleOutputStreamOperator<UserEvent> beanStream = jsonStream.map(value -> JSON.parseObject(value, UserEvent.class));

        // 将数据keyBy(用户)
        KeyedStream<UserEvent, Long> keyByStream = beanStream.keyBy(UserEvent::getUid);


        /* *
         * 用 mysql-cdc连接器（source），去抓取 规则元数据信息表中的一切变化操作（增、删、改）
         */
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("doitedu")
                .port(3306)
                .databaseList("doit41") // set captured database, If you need to synchronize the whole database, Please set tableList to ".*".
                .tableList("doit41.rule_meta") // set captured table
                .username("root")
                .password("root")
                .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
                .build();


        DataStreamSource<String> ruleMetaCdcStream = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "rule_meta_source");

        SingleOutputStreamOperator<RuleMetaBean> ruleMetaBeanStream = ruleMetaCdcStream.map(new MapFunction<String, RuleMetaBean>() {
            @Override
            public RuleMetaBean map(String cdcJson) throws Exception {

                RuleMetaBean ruleMetaBean = new RuleMetaBean();

                // 提取op
                JSONObject jsonObject = JSON.parseObject(cdcJson);
                String op = jsonObject.getString("op");
                ruleMetaBean.setOp(op);

                // 提取表数据
                JSONObject dataObject;
                if (op.equals("d")) {
                    dataObject = jsonObject.getJSONObject("before");
                } else {
                    dataObject = jsonObject.getJSONObject("after");
                }

                // 提取规则id
                /**
                 *     "rule_id": "rule-1-002",
                 *     "model_id": "model-1",
                 *     "rule_param_json": "{bbbbbbbbbbbbb}",
                 *     "rule_status": 0
                 */
                String ruleId = dataObject.getString("rule_id");
                String modelId = dataObject.getString("model_id");
                String ruleParamJson = dataObject.getString("rule_param_json");
                int ruleStatus = dataObject.getIntValue("rule_status");
                Long webCalcEndtime = dataObject.getLong("web_calc_endtime");


                // 提取规则的预圈选人群bitmap的序列化字节
                byte[] preSelectUsersBitmapBytes = dataObject.getBytes("pre_select_users");
                Roaring64Bitmap bitmap = Roaring64Bitmap.bitmapOf();
                if (preSelectUsersBitmapBytes != null) {
                    bitmap.deserialize(ByteBuffer.wrap(preSelectUsersBitmapBytes));
                }

                // 提取规则的运算机代码
                String calculatorCode = dataObject.getString("calculator_code");


                // 把提取到的信息，填充到ruleMetaBean对象
                ruleMetaBean.setRule_id(ruleId);
                ruleMetaBean.setModel_id(modelId);
                ruleMetaBean.setRule_param_json(ruleParamJson);
                ruleMetaBean.setRule_status(ruleStatus);
                ruleMetaBean.setPreSelectUsersBitmap(bitmap);
                ruleMetaBean.setCalculatorCode(calculatorCode);
                ruleMetaBean.setWebCalcEndTime(webCalcEndtime);

                return ruleMetaBean;
            }
        });

        // 广播规则元数据流
        MapStateDescriptor<String, RuleMetaBean> bcStateDesc = new MapStateDescriptor<>("bc_state", String.class, RuleMetaBean.class);
        BroadcastStream<RuleMetaBean> metaBeanBroadcastStream = ruleMetaBeanStream.broadcast(bcStateDesc);

        // 连接 行为事件keyBy主流  和   规则元数据广播流
        SingleOutputStreamOperator<String> messageStream =
                keyByStream
                        .connect(metaBeanBroadcastStream)
                        .process(new KeyedBroadcastProcessFunction<Long, UserEvent, RuleMetaBean, String>() {

                            ConcurrentHashMap<String, RuleCalculator> calculatorPool = new ConcurrentHashMap<>();


                            GroovyClassLoader groovyClassLoader;
                            ListState<UserEvent> eventListState;
                            ValueState<HashSet<String>> oldRuleIdsState;
                            MapState<String, RuleCalculator> calculatorMapState;

                            @Override
                            public void open(Configuration parameters) throws Exception {
                                groovyClassLoader = new GroovyClassLoader();

                                // 开辟一个缓存最近用户行为数据的状态，用于缓存最近10s的数据
                                ListStateDescriptor<UserEvent> desc = new ListStateDescriptor<>("recent_events", UserEvent.class);
                                desc.enableTimeToLive(StateTtlConfig.newBuilder(Time.seconds(10)).build());

                                eventListState = getRuntimeContext().getListState(desc);

                                // 开辟一个状态，用于记录已存在过的运算机对应的规则id
                                oldRuleIdsState = getRuntimeContext().getState(
                                        new ValueStateDescriptor<HashSet<String>>(
                                                "old_rule_ids",
                                                TypeInformation.of(new TypeHint<HashSet<String>>() {
                                                })));

                                // 申请一个MapState用于存放运行中的运算机对象
                                calculatorMapState = getRuntimeContext().getMapState(new MapStateDescriptor<String, RuleCalculator>("calculator_pool", String.class, RuleCalculator.class));

                            }

                            @Override
                            public void processElement(UserEvent userEvent, KeyedBroadcastProcessFunction<Long, UserEvent, RuleMetaBean, String>.ReadOnlyContext ctx, Collector<String> out) throws Exception {


                                if (userEvent.getEvent_id().equals("error") && RandomUtils.nextInt(1, 11) % 4 == -0) {
                                    throw new RuntimeException("哈哈哈哈，就让你错");
                                }


                                /**
                                 * 故障后的恢复流程
                                 */
                                if (calculatorPool.isEmpty()) {
                                    ReadOnlyBroadcastState<String, RuleMetaBean> ruleMetaBcState = ctx.getBroadcastState(bcStateDesc);
                                    // 遍历备份的规则元数据，逐一构建运算机对象，并初始化，并放入运算机池
                                    for (Map.Entry<String, RuleMetaBean> entry : ruleMetaBcState.immutableEntries()) {
                                        String ruleId = entry.getKey();
                                        RuleMetaBean metaBean = entry.getValue();
                                        // 构建并初始化运算机
                                        RuleCalculator ruleCalculator = buildRuleCalculator(metaBean);
                                        // 放入运算机池
                                        calculatorPool.put(ruleId, ruleCalculator);
                                    }
                                }


                                /**
                                 * 正常数据处理流程
                                 */
                                HashSet<String> oldRules = oldRuleIdsState.value();

                                if (oldRules == null) oldRules = new HashSet<>();

                                Set<Map.Entry<String, RuleCalculator>> entries = calculatorPool.entrySet();

                                // keyedState行不通，因为processBroadcastElement方法中不具备keyed上下文，无法操作keyedState
                                // Iterable<Map.Entry<String, RuleCalculator>> entries = calculatorMapState.entries();

                                //  广播状态也行不通，因为运算机对象本身就持有状态，状态套状态，序列化失败
                                //Iterable<Map.Entry<String, RuleCalculator>> entries = bcState.immutableEntries();

                                // 遍历每一个运算机，调用它来处理当前的数据
                                for (Map.Entry<String, RuleCalculator> entry : entries) {
                                    String ruleId = entry.getKey();
                                    RuleCalculator calculator = entry.getValue();

                                    // 如果是一个老运算机，则直接让它处理当前收到的数据
                                    if (oldRules.contains(ruleId)) {
                                        calculator.calc(userEvent, out);

                                    } else {
                                        // 如果是一个新运算机，则应该先把缓存中的数据，逐条传给它处理
                                        for (UserEvent oldEvent : eventListState.get()) {
                                            calculator.calc(oldEvent, out);
                                        }

                                        // 缓存中的数据处理完成，再把新收到的这一条数据交给它处理
                                        calculator.calc(userEvent, out);

                                        // 将本运算机放入oldRules集合
                                        oldRules.add(ruleId);
                                        oldRuleIdsState.update(oldRules);

                                    }

                                }

                                // 将本次收到的这条数据放入缓存
                                eventListState.add(userEvent);
                            }

                            @Override
                            public void processBroadcastElement(RuleMetaBean ruleMetaBean, KeyedBroadcastProcessFunction<Long, UserEvent, RuleMetaBean, String>.Context ctx, Collector<String> out) throws Exception {

                                // 获取广播状态（用于备份规则元数据）
                                BroadcastState<String, RuleMetaBean> ruleMetaBcState = ctx.getBroadcastState(bcStateDesc);

                                /**
                                 * 故障后的运算机自动恢复逻辑
                                 */
                                if (calculatorPool.isEmpty()) {
                                    // 遍历备份的规则元数据，逐一构建运算机对象，并初始化，并放入运算机池
                                    for (Map.Entry<String, RuleMetaBean> entry : ruleMetaBcState.entries()) {
                                        String ruleId = entry.getKey();
                                        RuleMetaBean metaBean = entry.getValue();
                                        // 构建并初始化运算机
                                        RuleCalculator ruleCalculator = buildRuleCalculator(metaBean);
                                        // 放入运算机池
                                        calculatorPool.put(ruleId, ruleCalculator);

                                    }
                                }


                                /**
                                 * 正常情况下的规则上下线逻辑
                                 */
                                // 增加规则
                                if (
                                        (
                                                ruleMetaBean.getOp().equals("c")
                                                        || ruleMetaBean.getOp().equals("r")
                                                        || ruleMetaBean.getOp().equals("u")
                                        )
                                                && ruleMetaBean.getRule_status() == 1
                                ) {

                                    // 构建并初始化运算机对象
                                    RuleCalculator calculator = buildRuleCalculator(ruleMetaBean);

                                    // 放入运算机池
                                    calculatorPool.put(ruleMetaBean.getRule_id(), calculator);
                                    // 将元数据备份到广播状态
                                    ruleMetaBcState.put(ruleMetaBean.getRule_id(), ruleMetaBean);

                                    // 根本行不通，因为在processBroadcastElement中不能操作keyedState
                                    // calculatorMapState.put(ruleMetaBean.getRule_id(), calculator);
                                    // bcState.put(ruleMetaBean.getRule_id(), calculator);

                                    log.warn("增加了一个规则运算机，规则id为：{}", ruleMetaBean.getRule_id());
                                }

                                // 删除、下线规则
                                if (ruleMetaBean.getOp().equals("d") || (ruleMetaBean.getOp().equals("u") && ruleMetaBean.getRule_status() == 0)) {

                                    calculatorPool.remove(ruleMetaBean.getRule_id());
                                    // calculatorMapState.remove(ruleMetaBean.getRule_id());
                                    // bcState.remove(ruleMetaBean.getRule_id());

                                    // 从元数据备份广播状态中移除该规则的元数据
                                    ruleMetaBcState.remove(ruleMetaBean.getRule_id());


                                    log.warn("下线了一个规则运算机，规则id为：{}", ruleMetaBean.getRule_id());
                                }
                            }

                            /**
                             * 根据规则元数据构建运算机的工具方法
                             * @param ruleMetaBean
                             * @return RuleCalculator
                             */
                            private RuleCalculator buildRuleCalculator(RuleMetaBean ruleMetaBean) throws InstantiationException, IllegalAccessException, IOException {
                                // 根据规则运算机源代码，构建运算机对象
                                Class aClass = groovyClassLoader.parseClass(ruleMetaBean.getCalculatorCode());
                                RuleCalculator calculator = (RuleCalculator) aClass.newInstance();


                                // 初始化运算机对象
                                calculator.init(getRuntimeContext(),
                                        ruleMetaBean.getRule_param_json(),
                                        ruleMetaBean.getPreSelectUsersBitmap(),
                                        ruleMetaBean.getWebCalcEndTime());
                                return calculator;
                            }
                        });


        messageStream.print();

        env.execute();

    }

}
